package blocksim.chunking;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.List;

import blocksim.hash.HashStrategy;
import blocksim.hash.HashStrategyFactory;
import blocksim.util.ArrayUtils;

/**
 * Content based block boundary detection (chunking) algorithm as explained in
 * [1]:
 * 
 * "CbCH scans the file using a ‘window’ of m bytes and, for each position of
 * the window, computes a hash of the corresponding string. A chunk boundary is
 * declared if the lowest k bits of the hash are all zero. Then, identification
 * of chunk similarity proceeds as above, based on chunk hashes. Statistically,
 * k, the number of bits of the hash compared to zero allows controlling the
 * average chunk size, while m, the window size, and p, the number of bytes the
 * window is advanced every time, allow controlling the variation in chunk sizes
 * and, additionally, influence the chunk size."
 * 
 * [1] Samer Al-Kiswany, Matei Ripeanu, Sudharshan Vazhkudai, Abdullah Gharaibeh
 * "stdchk: A Checkpoint Storage System for Desktop Grid Computing," 28th
 * International Conference on Distributed Computing Systems (ICDCS), 2008.
 * 
 * @author marcus
 */
public class ContentBasedChunking implements ChunkingStrategy {

  public static int M_DEFAULT = 20;
  public static int K_DEFAULT = 14;
  public static int P_DEFAULT = 1;
  public static String BOUNDARY_HASHF_DEFAULT = "SHA";
  public static long MIN_CHUNKSIZE_DEFAULT = 1;
  public static long MAX_CHUNKSIZE_DEFAULT = Long.MAX_VALUE;
  public static int BUFFER_SIZE = 16384;

  private int m;
  private int k;
  private int p;
  private long minChunkSize;
  private long maxChunkSize;
  private HashStrategy boundHashSt;
  //private DescriptiveStatistics chunkSizeStats;

  public ContentBasedChunking() throws NoSuchAlgorithmException {
    this(M_DEFAULT, K_DEFAULT, P_DEFAULT, MIN_CHUNKSIZE_DEFAULT, MAX_CHUNKSIZE_DEFAULT,
         BOUNDARY_HASHF_DEFAULT);
  }

  /**
   * Content based chunking class constructor. For more details, refer to [1].
   * 
   * @param m
   *          Window size (bytes)
   * @param k
   *          Number of verification bits
   * @param p
   *          Number of bytes to be skipped at each iteration
   * @param minChunkSize
   * @param maxChunkSize
   * @param boundaryHash
   *          Hash strategy name supported by {@link HashStrategyFactory}
   * @throws NoSuchAlgorithmException
   */
  public ContentBasedChunking(int m, int k, int p, long minChunkSize, long maxChunkSize,
      String boundaryHash) throws NoSuchAlgorithmException {
    this.m = m;
    this.k = k;
    this.p = p;
    this.minChunkSize = Math.max(minChunkSize, m);
    this.maxChunkSize = Math.max(maxChunkSize, m);
    this.boundHashSt = HashStrategyFactory.getInstance(boundaryHash, m);
    //this.chunkSizeStats = new DescriptiveStatistics();
  }

  @Override
  public List<Long> detectBoundaries(String fileName) throws IOException {
    ArrayList<Long> boundaries = new ArrayList<Long>();
    BitSet mask = new BitSet(k);
    mask.set(0, k);
    byte[] initWindow = new byte[m];
    byte[] slideBytes = new byte[p];
    byte[] buffer;
    boolean isBoundary = false;
    boolean eof = false;
    int bytesRead = 0;
    long chunkBegin = 0;
    long chunkEnd = 0;
    RandomAccessFile in = new RandomAccessFile(fileName, "r");

    do {
      if (chunkEnd == chunkBegin) {
        if (minChunkSize > m) {
          // Skip data stream until the minimum chunk size
          chunkEnd += in.skipBytes((int) (minChunkSize - m));
        }
        // Read entire window size (m) when starting a new chunk
        buffer = initWindow;
      } else {
        // Read only the next bytes of the sliding window
        buffer = slideBytes;
      }

      bytesRead = in.read(buffer);
      chunkEnd += Math.max(0, bytesRead);
      if (bytesRead < Math.min(p, m)) {
        eof = true;
      } else {
        long chunkSize = chunkEnd - chunkBegin;
        chunkSize = Math.min(chunkSize, maxChunkSize);
        if (chunkSize == maxChunkSize) {
          isBoundary = true;
          chunkEnd = chunkBegin + chunkSize;
        } else {
          // Update window with new bytes read
          boundHashSt.pushBytes(buffer, 0, bytesRead);
          isBoundary = checkBoundary(mask, buffer, bytesRead);
        }
        if (isBoundary) {
          boundaries.add(chunkEnd);
          //chunkSizeStats.addValue(chunkSize);
          // System.err.println(minchunkSize + "\t" + chunkBegin + "\t" +
          // chunkSize);
          chunkBegin = chunkEnd;
        }
      }
    } while (!eof);

    if (chunkEnd > chunkBegin) {
      // Add the last byte as boundary if not added yet
      boundaries.add(chunkEnd);
      //chunkSizeStats.addValue(chunkEnd - chunkBegin);
    }

    in.close();
    return boundaries;
  }

  /**
   * Check if the last k bits of the data window are equal to the mask, where k
   * is the number of bits in the mask
   * 
   * @param mask
   *          Bit mask of size k
   * @param buffer
   *          Data window
   * @param bytesRead
   *          Number of bytes to use from data window
   * @return
   */
  private boolean checkBoundary(BitSet mask, byte[] buffer, int bytesRead) {
    byte[] hash = boundHashSt.hash();
    BitSet hashBitSet = ArrayUtils.byteArrayToBitSet(hash);
    hashBitSet.and(mask);
    if (hashBitSet.cardinality() == 0) {
      return true;
    }
    return false;
  }

  @Override
  public String getName() {
    return boundHashSt.getName();
  }

  //@Override
  //public DescriptiveStatistics getBlockSizeStats() {
    //return chunkSizeStats;
  //}
}
